package com.lagou.client.handler;

import com.lagou.client.util.ZkClientUtil;
import com.lagou.invoke.RpcRequest;
import com.lagou.invoke.RpcResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.Callable;

/**
 * @author Kirito
 * @description:
 * @date 2020/05/26
 */
public class RpcClientHandler extends ChannelInboundHandlerAdapter implements Callable<Object> {

    private ChannelHandlerContext context;

    private RpcResponse result;

    private RpcRequest request;

    private long responseTime;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.context = ctx;
    }

    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        RpcResponse response = (RpcResponse) msg;
        // 计算响应时间
        long use = System.currentTimeMillis() - response.getStartTime();
        // 将响应时间作为数据存储到ZK的服务节点
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        int port = ipSocket.getPort();
        String host = ipSocket.getHostString();
        String address = host + ":" + port;
        ZkClientUtil.setResponse(address, use);
        System.out.println(address + " response time is: " + use);
        this.result = response;
        // 收到响应后唤醒等待的线程
        notify();
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        this.responseTime = System.currentTimeMillis();
    }

    public synchronized Object call() throws Exception {
        context.writeAndFlush(request).sync();
        // 阻塞线程，等待响应结果
        wait();
        return result.getData();
    }

    public void setRequest(RpcRequest request) {
        this.request = request;
    }
}
